//
// Created by zhfayuan on 2021/6/15.
//

#include "GBGwChannel.h"
#include "MediaType.h"
#include <iostream>
static gbt::GbtProxyAgent& getAgent()
{
    return gbt::GbtProxyAgent::instance();
}

GBGwChannel::GBGwChannel(GBGwChannelHandler* handle,std::string name,std::string parent,std::string dev,uint32_t index)
:_live_handler(handle),_pktRun(false),_pktAbort(false),_pktThread(nullptr),_online(false),_index(index),_device(nullptr)
{

     char deviceID[GBT_UNICODE_LENGTH + 1] = {0};
     char prarentID[GBT_UNICODE_LENGTH + 1] = {0};
     memcpy(deviceID,dev.c_str(),dev.size());
     memcpy(prarentID,parent.c_str(),parent.size());
    //gbtCodeMakeSubEncoderDevice(parent.c_str(), index, deviceID, GBT_UNICODE_LENGTH);

    memset(&_gbtCatalog,0,sizeof(_gbtCatalog));
    strcpy(_gbtCatalog.name,name.c_str());
    strcpy(_gbtCatalog.deviceID,deviceID);
    strcpy(_gbtCatalog.parentID,prarentID);
    strcpy(_gbtCatalog.address, "Shanghai");
    strcpy(_gbtCatalog.manufacturer, "sh-easy");
    strcpy(_gbtCatalog.model, "EasyGbSvr");
    strcpy(_gbtCatalog.owner, "EasyGbSvr");
    strcpy(_gbtCatalog.civilCode, "435200");
    strcpy(_gbtCatalog.block, "435200");
    
    _gbtCatalog.longitude = 120.0165;
    _gbtCatalog.latitude = 31.8138;
    _gbtCatalog.status = GBT_STATUS_OFF;
    _gbtCatalog.parental  = 0;
    
    _aac_decoder = std::make_shared<AacDecoder>();
    _aac_encoder = std::make_shared<AacEncoder>();
    _audio_resampler = std::make_shared<AudioResampler>();

#if 1
    _device = getAgent().createDevice(index, _gbtCatalog.deviceID);
    if(_device){
        _device->setStatus(false);
        _device->setSink(this);
        _device->open(_gbtCatalog, nullptr);
    }
#endif
#if 1
    int rc = -1;
    if(_device)
        rc = _device->open(_gbtCatalog, nullptr);
    if (rc == 0)
    {
        std::cout << "GBGwChannel::open rc:"<< rc << std::endl;
    }
#endif
}

GBGwChannel::~GBGwChannel()
{
    offline();
    _aac_decoder = nullptr;
    _aac_encoder = nullptr;
    _audio_resampler = nullptr;
}


void GBGwChannel::pktThreadEntry()
{
    MFormat format = {0};
    bool v_checked = false;
    bool a_checked = false;
    bool check_expired = false;
    std::shared_ptr<RawPacket> raw;
    uint32_t expire = 0;

    do {
        {
            std::unique_lock<std::mutex> lck(_pktLock);
            if (_pktAbort)
                break;
            if(_pktQue.size() <= 0){
                _pktCond.wait(lck);
                continue;
            }
            raw = _pktQue.front();
            _pktQue.pop_front();
        }

        if(check_expired){
            MPacket packet;
            packet.type = raw->type;
            packet.pts = raw->pts;
            packet.size = raw->data.size();
            packet.data = raw->data.data();
            packet.flags = raw->key?1:0;

            if(packet.type == MTYPE_AUDIO){

#ifdef AAC_RESAMPLE
                auto aac_encode_out = [&](std::shared_ptr<std::vector<char>>&aac,uint32_t pts_ms){
                    packet.size = aac->size();
                    packet.data = (uint8_t*)aac->data();
#ifdef NO_ADTS_HEADER
                    packet.size = aac->size() - 7;
                    packet.data = (uint8_t*)aac->data() + 7;
                    _device->writePacket(packet);
#else
                    _device->writePacket(packet);
#endif
                    std::cout << "_device->getDeviceID() :"<<_device->getDeviceID() << ",encode out" << std::endl;
                };

                auto audio_resample_out = [&](std::shared_ptr<std::vector<char>>&out,int32_t rate,int32_t channel){
                    _aac_encoder->initialize(rate,16,channel);
                    _aac_encoder->encode(out,raw->pts,aac_encode_out);
                };

                auto aac_decode_out = [&](std::shared_ptr<std::vector<char>>&pcm,int32_t sample_rate,int32_t channel){
                    if(sample_rate == 44100 && channel == 2){
                        _audio_channel == channel;
                        _audio_sample_rate = sample_rate;
                        std::cout << "_device->getDeviceID() :"<<_device->getDeviceID() << ",decode out" << std::endl;
#ifdef NO_ADTS_HEADER
                        packet.data = raw->data.data() + 7;
                        packet.size = raw->data.size() - 7;
                        _device->writePacket(packet);
#else
                        _device->writePacket(packet);
#endif
                        return;
                    } else if(_audio_sample_rate != sample_rate || _audio_channel != channel){
                        _audio_resampler.reset( new AudioResampler());
                        _audio_channel = channel;
                        _audio_sample_rate = sample_rate;
                        std::cout << "_device->getDeviceID() :"<<_device->getDeviceID() << ",new resample for sample_rate:%d " << sample_rate << ",ch:"<< channel << std::endl;
                    }

                    _audio_resampler->initialize(sample_rate,44100,2);
                    if(_audio_channel == 2 ) {
                        std::cout << "_device->getDeviceID() :"<<_device->getDeviceID() << ",channel:" << _audio_channel << "!=2"  << std::endl;
                        _audio_resampler->resample(pcm,audio_resample_out);
                    } else if( 1 == _audio_channel ) {
                        uint16_t  *src,*dest;
                        auto pcm_x2  = std::make_shared<std::vector<char>>();
                        pcm_x2->resize(pcm->size() * 2);
                        src = (uint16_t*)pcm->data();
                        dest = (uint16_t*)pcm_x2->data();
                        for (size_t i = 0;i  i < (pcm->size() >> 1) ; ++i)
                        {
                            *dest =  *src; dest++;
                            *dest =  *src; dest++; src++;
                        }
                        _audio_resampler->resample(pcm_x2,audio_resample_out);
                    }
                };
                auto aac = std::make_shared<std::vector<char>>();
                aac->assign(raw->data.data(),raw->data.data() + raw->data.size() );
                _aac_decoder->decode(aac,aac_decode_out);
#else

#ifdef NO_ADTS_HEADER
                packet.size = raw->data.size() -7;
                packet.data = raw->data.data() + 7;
                _device->writePacket(packet);
#else
                _device->writePacket(packet);
#endif

#endif


            }  else if(packet.type == MTYPE_VIDEO){
#ifdef DROP_ALL_HEVC
                if(raw->hevc)
                    continue;
#endif

                //std::cout << "GBGwChannel("<< _gbtCatalog.deviceID  <<") size:"<<  packet.size <<",flags:"<< packet.flags << std::endl;
            
                _device->writePacket(packet);
            }
        } else {

            uint32_t 
            now = time(nullptr);

            if(!expire) {
                expire = now + 5;
                memset(&format,0,sizeof (format));
                format.clockRate = 1000000;
                format.audioRate = 1000000;
            }

            if(now < expire){
                if((raw->type == MTYPE_VIDEO) &&
#ifdef DROP_ALL_HEVC
                        !(raw->hevc) &&
#endif
                        !v_checked  ){
                    format.codec = raw->hevc? av::MEDIA_CODEC_HEVC:av::MEDIA_CODEC_H264;
                    format.width = 0;
                    format.height = 0;
                    v_checked = true;
                }else if((raw->type == MTYPE_AUDIO) && !a_checked){
                    format.audioCodec = av::MEDIA_CODEC_AAC;
                    a_checked = true;
                }

                if(a_checked && v_checked){
                    check_expired = true;
                    if(_device)
                    _device->setFormat(format);
                    std::cout << "GBGwChannel("<< _gbtCatalog.deviceID  <<") check_expired at:"<< check_expired <<",now:"<< now <<",v_checked:"<< v_checked <<",a_checked:"<< a_checked << std::endl;
                }
                
            } else {
                check_expired = true;
                if(_device)
                    _device->setFormat(format);
                std::cout << "GBGwChannel("<< _gbtCatalog.deviceID  <<") check_expired at:"<< check_expired <<",now:"<< now <<",v_checked:"<< v_checked <<",a_checked:"<< a_checked << std::endl;
            }
        }
    }while(!_pktAbort);
}

void GBGwChannel::online()
{
   //std::unique_lock<std::mutex> lck(_opLock);
    std::cout << "GBGwChannel("<< _gbtCatalog.deviceID  << ") online "<<  ",device:" << _device<< std::endl;
    _last_v_ts = _last_a_ts = 0;
    #if 0
    {
        std::unique_lock<std::mutex> lck(_pktLock);
        if(_device)
            return;
        _device = getAgent().createDevice(_index, _gbtCatalog.deviceID);
        if(_device){
            _device->setSink(this);
            _gbtCatalog.status = GBT_STATUS_ON;
            _device->open(_gbtCatalog, nullptr);
            _device->setStatus(true);
        }
    }
    #else
    {
         std::unique_lock<std::mutex> lck(_pktLock);
        _gbtCatalog.status = GBT_STATUS_ON;
        _device->setStatus(true);
    }
    #endif


    {
        auto pkt_f =[&](){
            {
                std::unique_lock<std::mutex> lck0(_pktLock);
                _pktCond.notify_all();
                _pktRun = true;
            }
            pktThreadEntry();
            {
                std::unique_lock<std::mutex> lck0(_pktLock);
                _pktCond.notify_all();
                _pktRun = false;
            }
        };
        
        //启动消息处理
        std::unique_lock<std::mutex> lck(_pktLock);
        if(_pktRun)
            return;
        _pktAbort = false;
        _pktThread = std::make_shared<std::thread>(pkt_f);
        _pktCond.wait(lck);
    }
    _online = true;

}

void GBGwChannel::offline()
{
   //std::unique_lock<std::mutex> lck(_opLock);
   #if 0
    {
        std::unique_lock<std::mutex> lck(_pktLock);
        if(!_device)
            return;
    }
    #endif
    {
        {
            std::unique_lock<std::mutex> lck(_pktLock);
            if (_pktRun && _pktThread) {
                _pktAbort = true;
                _pktCond.notify_all();
            }
        }
        if (_pktThread)
        {
            _pktThread->join();
            _pktThread = nullptr;
        }
    }

    {
        std::unique_lock<std::mutex> lck(_pktLock);
        _pktQue.clear();

        if(_device){
            _device->setStatus(false);
            _gbtCatalog.status = GBT_STATUS_OFF;
            #if 0
            _device->close();
            _device->setSink(nullptr);
            getAgent().destroyDevice(_device);
            _device = nullptr;
            #endif
        }
    }

    _online = false;
    std::cout << "GBGwChannel("<< _gbtCatalog.deviceID  << ") offline "<<  ",device:" << _device<< std::endl;
}

GBT_Catalog & GBGwChannel::getCatalog()
{

    std::cout << "getCatalog("<< _gbtCatalog.deviceID  << ") online "<<  ",parental:" << _gbtCatalog.parental << std::endl;
    return  _gbtCatalog;
}

void GBGwChannel::writeAudio(char *data, uint32_t length, uint64_t  pts)
{
    //static  int32_t print_cnt = 0;
    if(!_online)
        return;

    //print_cnt ++;

    //if( !(print_cnt & 0xF)){
    //    std::cout << "GBGwChannel a("<< _gbtCatalog.deviceID  << ") "<<  ",pts:" << pts << std::endl;
    //}

    if(pts < _last_a_ts){
        std::cout << "GBGwChannel ------------------------>a("<< _gbtCatalog.deviceID  << ") "<<  ",pts:" << pts << "->>"<< _last_a_ts<< std::endl;
    }
    _last_a_ts = pts;


    auto raw = std::make_shared<RawPacket>();
    raw->data.assign(data,data+length);
    raw->pts = pts;
    raw->hevc = false;
    raw->key = true;
    raw->type = MTYPE_AUDIO;
    {
        std::unique_lock<std::mutex> lck(_pktLock);
        if(!_device)
            return;
        if (_pktRun && _pktThread) {
            _pktQue.push_back(raw);
        }
        _pktCond.notify_all();
    }
}

void GBGwChannel::writeVideo(char *data, uint32_t length, uint64_t  pts,bool key,bool hevc)
{
    if(!_online)
        return;

    //static int32_t print_cnt = 0;

    //print_cnt ++;
    //if( !(print_cnt & 0xF)){
    //    std::cout << "GBGwChannel v("<< _gbtCatalog.deviceID  << ") "<<  ",pts:" << pts << std::endl;
    //}


    if(pts < _last_v_ts){
        std::cout << "GBGwChannel ------------------------>v("<< _gbtCatalog.deviceID  << ") "<<  ",pts:" << pts << "->>"<< _last_v_ts<< std::endl;
    }
    _last_v_ts = pts;

    auto raw = std::make_shared<RawPacket>();
    raw->data.assign(data,data+length);
    raw->pts = pts;
    raw->hevc = hevc;
    raw->key = key;
    raw->type = MTYPE_VIDEO;

    {
        std::unique_lock<std::mutex> lck(_pktLock);
        if(!_device)
            return;
        if (_pktRun && _pktThread) {
            _pktQue.push_back(raw);
        }
        _pktCond.notify_all();
    }
}

void GBGwChannel::onBeginPlay(gbt::GbtMediaDevice *device, const char *peerID)
{
    if(!_live_handler)
        return;
    _live_handler->onBeginPlay(_index);
}

void GBGwChannel::onEndPlay(gbt::GbtMediaDevice *device, const char *peerID)
{
   if(!_live_handler)
        return;
    _live_handler->onEndPlay(_index);
}
